package org.example.listener;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import io.lettuce.core.cluster.api.reactive.RedisAdvancedClusterReactiveCommands;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.Duration;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RedisSubscriber {

    // 配置哨兵节点（192.168.0.101 为第一个哨兵，mymaster这是 Redis 哨兵配置中的主节点名称（可以在 sentinel.conf 中找到），客户端需要根据这个名称从哨兵节点获取主节点地址。
    /*RedisURI redisUri = RedisURI.Builder.sentinel("192.168.0.101", 26379, "mymaster")  // 主节点名称为 "mymaster" 配置哨兵节点的地址和主节点的名称（masterName），Redis 哨兵将会自动返回主节点地址。
            .withSentinel("192.168.0.102", 26379)  // 第二个哨兵节点
            .withSentinel("192.168.0.103", 26379)  // 第三个哨兵节点
            .withPassword("your_password")         // 设置 Redis 认证密码
            .withDatabase(0)                       // 使用 Redis 数据库 0
            .build();
    // 创建 Redis 客户端
    RedisClient redisClient = RedisClient.create(redisUri);
    // 建立连接
    StatefulRedisConnection<String, String> connection = redisClient.connect();
    // 执行命令
    connection.sync().set("key", "value");
    System.out.println("Get key: " + connection.sync().get("key"));
    // 关闭连接
    connection.close();
    redisClient.shutdown();*/

    public void test(){
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(100);  // 最大连接数
        poolConfig.setMaxIdle(10);    // 最大空闲连接数
        poolConfig.setMinIdle(5);     // 最小空闲连接数

        RedisClusterClient clusterClient = RedisClusterClient.create(Arrays.asList(
                RedisURI.Builder.redis("127.0.0.1", 6379).withPassword("your_password".toCharArray()).build(),
                RedisURI.Builder.redis("127.0.0.2", 6379).withPassword("your_password".toCharArray()).build(),
                RedisURI.Builder.redis("127.0.0.3", 6379).withPassword("your_password".toCharArray()).build()
        ));
        // 获取 Redis 集群连接
        try(StatefulRedisClusterConnection<String, String> connection = clusterClient.connect()) {
            // 获取同步命令接口
            RedisAdvancedClusterCommands<String, String> commands = connection.sync();
            // 执行 Redis 命令
            commands.set("key", "value");
            String value = commands.get("key");
            System.out.println("Key: " + value);
            // 异步连接允许你以非阻塞方式发送 Redis 命令，通过 async() 方法获取异步接口。
            RedisAdvancedClusterAsyncCommands<String, String> asyncCommands = connection.async();
            asyncCommands.set("key", "value").thenAccept(System.out::println);
            // 获取反应式（Reactive）连接
            RedisAdvancedClusterReactiveCommands<String, String> reactiveCommands = connection.reactive();
            reactiveCommands.get("key").subscribe(value1 -> System.out.println("Reactive Key: " + value1));

            /*. Redis 事务的基本工作流程
            MULTI：开启事务，将后续的命令放入队列。
            EXEC：执行事务，执行事务队列中的所有命令。
            DISCARD：取消事务，丢弃事务队列中的所有命令。
            WATCH（可选）：用于乐观锁机制，监控一个或多个键，在事务执行前检查是否被修改。
            */
            connection.close();  // 关闭连接
            clusterClient.shutdown();  // 关闭客户端
        }
    }

    // 单节点事物
    public void shiwu(){
        // 创建 RedisClient
        RedisClient redisClient = RedisClient.create("redis://localhost:6379");

        // 获取连接
        try (StatefulRedisConnection<String, String> connection = redisClient.connect()) {
            // 获取同步命令接口
            RedisCommands<String, String> commands = connection.sync();

            // 开启事务
            commands.multi();

            try {
                // 在事务中执行命令
                commands.set("key1", "value1");
                commands.set("key2", "value2");

                // 提交事务
                commands.exec();
                System.out.println("Transaction committed");

            } catch (Exception e) {
                // 发生错误时取消事务
                commands.discard();
                System.out.println("Transaction discarded due to error: " + e.getMessage());
            }
        }

        // 关闭 RedisClient
        redisClient.shutdown();
    }

    /*// 集群事物
    public void shiwu1(){
// 创建 RedisClusterClient
        RedisClusterClient clusterClient = RedisClusterClient.create("redis://localhost:6379");

        // 获取集群连接
        try (StatefulRedisClusterConnection<String, String> connection = clusterClient.connect()) {
            // 获取集群同步命令接口
            RedisAdvancedClusterCommands<String, String> commands = connection.sync();

            // 开启事务
            commands.multi();

            try {
                // 在事务中执行命令
                commands.set("key1", "value1");
                commands.set("key2", "value2");

                // 提交事务
                commands.exec();
                System.out.println("Transaction committed");

            } catch (Exception e) {
                // 发生错误时取消事务
                commands.discard();
                System.out.println("Transaction discarded due to error: " + e.getMessage());
            }
        }

        // 关闭 RedisClusterClient
        clusterClient.shutdown();
    }*/

    /*RedisClient redisClient = RedisClient.create(RedisURI.Builder.redis("127.0.0.1", 6379)
            .withPassword("your_password".toCharArray())
            //.withTimeout(Duration.ofSeconds(10))  // 设置超时时间
            .withDatabase(0)
            .build());*/
    private static final RedisClient redisClient = RedisClient.create("redis://localhost:6379");
    private static final StatefulRedisPubSubConnection<String, String> connection = redisClient.connectPubSub();
    private static final RedisPubSubAsyncCommands<String, String> async = connection.async();

    // 保存频道和客户端连接的映射
    private static final Map<String, Channel> subscribedChannels = new ConcurrentHashMap<>();

    public static void subscribeToChannel(String channel, Channel clientChannel) {
        // 订阅频道
        async.subscribe(channel);
        // 将客户端通道与频道关联起来
        subscribedChannels.put(channel, clientChannel);

        // 使用 RedisPubSubListener 监听消息
        connection.addListener(new RedisPubSubListener<String, String>() {
            @Override
            public void message(String channel, String message) {
                System.out.println("Received message on channel: " + channel);
                // 将消息推送给订阅了该频道的所有客户端
                Channel client = subscribedChannels.get(channel);
                if (client != null) {
                    client.writeAndFlush(new TextWebSocketFrame(message));
                }
            }

            @Override
            public void message(String pattern, String channel, String message) {
                // 用于处理基于模式的订阅消息
            }

            @Override
            public void subscribed(String channel, long count) {
                System.out.println("Subscribed to channel: " + channel);
            }

            @Override
            public void psubscribed(String pattern, long count) {
                // 用于处理基于模式的订阅
            }

            @Override
            public void unsubscribed(String channel, long count) {
                System.out.println("Unsubscribed from channel: " + channel);
            }

            @Override
            public void punsubscribed(String pattern, long count) {
                // 用于处理基于模式的取消订阅
            }
        });
    }
}


